package com.itcam.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

@SpringBootTest(classes = ConsumerApplication.class)
public class JavaConsumerDemo {

    // 定义主题
    public static final String TOPIC_NAME = "HelloKafka";

    @Test
    void consumerMsg() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.150.101:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group01");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // earliest：最开始第一次向一个topic发消息时，如果没有消费位移，那么此时从最早（最小）偏移量开始读取消息
        // latest：最开始第一次向一个topic发消息时，如果没有消费位移，那么此时从最新（最大）偏移量开始读取消息

        // 查看消费位移的命令如下：
        // kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group group01 --describe
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 创建消费者对象
        KafkaConsumer kafkaConsumer = new KafkaConsumer(properties);

        // 订阅主题
        kafkaConsumer.subscribe(Arrays.asList(TOPIC_NAME));

        while (true) {
            // 消费消息
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.printf("offset = %d, key = %s, value = %s%n",
                        consumerRecord.offset(), // offset表示消费位移
                        consumerRecord.key(),
                        consumerRecord.value());
            }
        }
    }
}